#include "Avalon/Network/TcpSocketReader.hpp"
#include <boost/bind.hpp>
#include "Avalon/IO/Buffer.hpp"
#include "Avalon/Network/SocketException.hpp"
#include "Avalon/Network/SocketType.hpp"
#include "Avalon/Network/TcpSocketConnection.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Network;
using namespace Avalon::Network::Detail;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::asio;
using namespace boost::asio::ip;
using namespace boost::posix_time;
using namespace std;

TcpSocketReader::~TcpSocketReader() {}

bool TcpSocketReader::IsAtEndOfDataSource() {
  return false;
}

Async<int>::Ptr TcpSocketReader::Read(Buffer* destination) {
  static const int READ_BUFFER_SIZE = 8192;
  return Read(destination, READ_BUFFER_SIZE);
}

Async<int>::Ptr TcpSocketReader::Read(void* destination, int size) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  AsyncOwner<int>::Ptr readResult = AsyncOwner<int>::Create();
  m_readResult = readResult;
  int readSize = size;
  m_socket->m_socket.async_read_some(buffer(destination, readSize),
    bind(&TcpSocketReader::HandleRead, shared_from_this(), destination,
    readSize, placeholders::error, placeholders::bytes_transferred));
  return readResult;
}

Async<int>::Ptr TcpSocketReader::Read(Buffer* destination, int size) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  AsyncOwner<int>::Ptr readResult = AsyncOwner<int>::Create();
  m_readResult = readResult;
  int readSize = size;
  destination->Reserve(readSize);
  m_socket->m_socket.async_read_some(buffer(destination->GetMutableData(),
    readSize), bind(&TcpSocketReader::HandleRead, shared_from_this(),
    destination->GetMutableData(), readSize, placeholders::error,
    placeholders::bytes_transferred));
  return readResult;
}

TcpSocketReader::TcpSocketReader(const boost::shared_ptr<SocketType>& socket)
    : m_socket(socket) {}

void TcpSocketReader::Abort() {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  if(m_readResult == NULL) {
    return;
  }
  AsyncOwner<int>::Ptr readResult = m_readResult;
  m_readResult.reset();
  readResult->SetException(IOException("Connection closed."));
}

void TcpSocketReader::HandleRead(void* destination, int readSize,
    const boost::system::error_code& error, size_t size) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  if(m_readResult == NULL) {
    return;
  }
  AsyncOwner<int>::Ptr readResult = m_readResult;
  m_readResult.reset();
  if(size != 0 && !error) {
    readResult->SetResult(static_cast<int>(size));
  } else if(error == asio::error::operation_aborted) {
    m_readResult = readResult;
    m_socket->m_socket.async_read_some(buffer(destination, readSize),
      bind(&TcpSocketReader::HandleRead, shared_from_this(), destination,
      readSize, placeholders::error, placeholders::bytes_transferred));
  } else {
    readResult->SetException(SocketException(error.value(),
      error.message().c_str()));
  }
}
